package operator.combine;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import util.RxJavaHelper;
import util.ThreadHelper;

class JoinOperator {

    public static void main(String[] args) {
        join();
    }

    /**
     * 第一个参数为要组合的数据，最后一个参数为组合方式
     * 中间两个为窗口定义，分别对应left和right的数据源，每一项数据都会通过窗口定义
     * 产生窗口期，只要两个窗口有交集(含头和尾)，就会产生数据
     */
    static void join() {
        Observable<String> left = Observable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
//                        Printer.print("left onNext aLong = " + aLong);
                    }
                })
                .map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Exception {
                        return "L" + aLong;
                    }
                })
                .take(5);


        Observable<String> right = Observable.interval(2, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
//                        Printer.print("Right onNext aLong = " + aLong);
                    }
                })
                .map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Exception {
                        return "R" + aLong;
                    }
                })
                .take(5);
        left.join(right, new Function<String, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(String s) throws Exception {
                return Observable.just(s).delay(1, TimeUnit.SECONDS);
            }
        }, new Function<String, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(String s) throws Exception {
                return Observable.just(s).delay(2, TimeUnit.SECONDS);
            }
        }, new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) throws Exception {
                return s + "-" + s2;
            }
        }).subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

}
